
----------------------------------------------------------------
--- Copyright：  迷你玩-海马
--- Description：data_cache缓存服务器连接到ssdb时使用(data_cache专用)
--- data_cache 使用ngx自带的长连接，连接到ssdb数据库
---
--- 业务服务器 ----> data_cache缓存服务器 --(ssdbMgrDc.lua)--> SSDB数据库
--- 修订1 第一代新版本 sema + resty_shared_lock版本
--- 修订2 20210201 第二代新版本  改动sema的队列
--- 修订3 20220208 第三代新版本  改动 try_connect + close 部分 更适合dc的单请求模式
----------------------------------------------------------------


local assert     = assert
local require    = require
local bit        = bit
local coroutine  = coroutine
local debug      = debug
local io         = io
local pairs      = pairs
local ipairs     = ipairs
local math       = math
local os         = os
local print      = print
local pcall      = pcall
local xpcall     = xpcall
local rawget     = rawget
local rawset     = rawset
local select     = select
local string     = string
local table      = table
local tonumber   = tonumber
local tostring	 = tostring
local error      = error
local type       = type
local unpack     = unpack
local setmetatable = setmetatable
local getmetatable = getmetatable
local ngx = ngx

require "resty.core.shdict"

local ngx_shared  = ngx.shared;
local string_sub  = string.sub
local string_gsub = string.gsub
local string_format = string.format

local log       = require("modules.commhm.log");
local ns_time   = require("modules.commhm.time");
local uu        = require("modules.commhm.utils");

local ns_ssdb   = require("modules.commhm.resty.ssdb");

local ns_semaphore  = require("ngx.semaphore")
--local ns_lock     = require("modules.commhm.resty.lock")

--local ns_env   = require( "modules.env.env" )


local table_insert = table.insert
local table_remove = table.remove
local ngx_sleep = ngx.sleep
local ngx_now   = ngx.now

local NGX_NULL  = "not_found"                  --ngx.null
local NGX_AUTH  = "authentication required."


-------------------------------------------------------------------------------
local __TJ_INTERVAL__      = 180

local __BUSY_BLOCK_LIMIT__   = 64       ---限流次数

local __SEMA_RESOURCES__       = 10         -- 最大并发赛道
local __SEMA_PROTECT_BREAK__   = (0-5*64)   -- 熔断保护值 (同时block mread read save)


---------------------


-- SSDB MGR类
-- _VERSION = '0.12.1'
local _M = {}
local mt = { __index = _M }


function _M.new(self, pre_, config_server_, config_select_ )
	log.debug( log.fast() and "call ssdbMgr:new" );
	local ins_ = {
		config_cache  = {};
		pre           = pre_;   --ma posting map

		select_mod    = (config_select_.select_mod or 10);  ---10分表  100分表

		config_server = config_server_;
		config_select = config_select_;

		sema_list = {};

		read_only_ssdb_block = 0;    ---忙的时候流控 无法mread

		last_time_report_tj        = uu.now() + __TJ_INTERVAL__;

		get_set_tj   = {};
		-- get_call
		-- get_cache
		-- get_cache_stale
		-- get_ssdb
		-- set_ssdb
		-- delay_ssdb
		-- set_re0
		-- set_re1

		mread_tj = {
			call  = 0,
			get   = 0,
			gets  = 0,
			set   = 0,
		};

		_lock_id   = (ngx.worker.id() + 1) * 1000;    --lock_id
	};
    return setmetatable(ins_, mt);
end


_M.final = function(self)
	self:do_report_tj()
end


---------------------------------------------------
---增加统计项
_M.set_get_set_tj = function(self, hash_, name_)

	if     ngx.ctx.m_params.act then
		hash_ = hash_ .. "_" .. ngx.ctx.m_params.act
	elseif ngx.ctx.m_params.cmd then
		hash_ = hash_ .. "_" .. ngx.ctx.m_params.cmd
	end

	if  not self.get_set_tj[hash_] then
		self.get_set_tj[hash_] = {
			get_call  = 0,
			get_cache = 0,
			get_cache_stale  = 0,
			get_ssdb = 0,
			set_ssdb = 0,
			delay_ssdb = 0,
			--set_re0 = 0,
			--set_re1 = 0,
		};
	end

	if  self.get_set_tj[hash_][ name_ ] then
		self.get_set_tj[hash_][ name_ ] = self.get_set_tj[hash_][ name_ ] + 1
	else
		log.error( "call set_get_set_tj, unknown name_=" .. name_ )
	end

end


---汇总打印统计项
_M.do_report_tj = function(self)
	local now_ = uu.now()
	if  now_ > self.last_time_report_tj then
		self.last_time_report_tj = now_ + __TJ_INTERVAL__;

		--开始打印统计
		for hash_, v1 in pairs(self.get_set_tj) do
			if  v1.get_call + v1.set_ssdb > 0 then
				log.day_list_short( "dc_ssdb", "cache_hit_ssdb", ngx.worker.id(), "get_set", self.pre,
					                hash_, uu.to_str(v1) )
			end
		end
		self.get_set_tj = {}

		if  self.mread_tj.call > 0 then
			log.day_list_short( "dc_ssdb", "cache_hit_ssdb", ngx.worker.id(), "mread", self.pre,
					uu.to_str( self.mread_tj ) )
			self.mread_tj = { call=0, get=0, gets=0, set=0 }
		end

	end
end


-------------------------------------------
--kvx 独立库
-------------------------------------------
_M.API_ssdb_set_kv = function( self, ssdb_name, key, value, et, err_call_back )
	if  log.fast() then
		log.debug( "call ssdbMgr.API_ssdb_set_kv, ssdb_name=" .. ssdb_name .. ", key=" .. key );
	end

	ngx.ctx.err_call_back = err_call_back;

	--优先写入share cache
	self:set_shared_cache_ssdb( ssdb_name, key .. "_kv", value, et )

	--lock 模式
	local lock_id_ = self:read_lock2( ssdb_name )
	if  lock_id_ == 0 then
		self:error_exit( "set_kv_lock_err" );
		return
	end

	local instance = self:try_connect(ssdb_name);
	local ok, err;
	if et and type(et)=="number" and et >= 0 then
		ok, err = instance:setx(key, "".. value, et);
		ok, err = instance:expire(key, et)
	else
		ok, err = instance:set(key, "" .. value);
	end

	self:try_close( instance, ssdb_name, err )
	self:read_unlock2(ssdb_name)

	if  not ok then
		log.day_list_short( "dc_ssdb",  "setx_fail", ssdb_name, key, err );
		self:error_exit( "setx err:" .. (err or "nil") );
	end

	self:set_get_set_tj( "kv", "set_ssdb" )

	return "OK"
end


_M.API_ssdb_get_kv = function( self, ssdb_name, key, err_call_back )
	if  log.fast() then
		log.debug( "call ssdbMgr.API_ssdb_get_kv, ssdb_name=" .. ssdb_name .. ", key=" .. key );
	end

	ngx.ctx.err_call_back = err_call_back;

	local ret2, stale_ = self:get_shared_cache_ssdb( ssdb_name, key .. "_kv" )
	if  ret2 then
		if  stale_ then
			--log.debug( log.fast() and "ignore stale_ ret2kv cache hit." )  --stale_也忽略
			self:set_get_set_tj( "kv", "get_cache_stale" )
		else
			--log.debug( log.fast() and "ret2kv cache hit." )
			self:set_get_set_tj( "kv", "get_cache" )
			return ret2   --cache命中
		end
	end

	self:set_get_set_tj( "kv", "get_call" )

	local ret = NGX_NULL;

	--lock 模式
	local lock_id_ = self:read_lock2( ssdb_name )
	if  lock_id_ == 0 then
		self:error_exit( "get_kv_lock_err" );
		return
	end

	local instance = self:try_connect(ssdb_name);
	local ret_table_, err = instance:get(key);

	self:try_close( instance, ssdb_name, err )
	self:read_unlock2(ssdb_name)

	--local ttl_ = instance:ttl( key );
	--uu.var_dump( ttl_ )

	if  err then
		log.day_list_short( "dc_ssdb", "get_fail", key, err );
		self:error_exit( "get err:" .. (err or "nil") );
		return;
	end

	uu.var_dump( ret_table_ );

	-- 处理返回值 返回是字符串或者NGX_NULL或NGX_AUTH
	if  type(ret_table_) == 'table' then
		ret = ret_table_[1];
		if  ret and type(ret) == 'string' then
			if      NGX_NULL == ret then
				--结果为空
				log.debug( log.fast() and "dc_ssdb ret not_found " .. key );
				return ret
			elseif  NGX_AUTH == ret then
				--没有权限
				self:error_exit( self.pre .. ssdb_name .. " ssdb " .. NGX_AUTH );
			else
				self:set_get_set_tj( "kv", "get_ssdb" )
				return ret
			end
		else
			--SSDB错误 一般是超时
			log.day_list_short( "dc_ssdb", "get_kv_fail", key, "ret_1_not_string", table.tostring(ret_table_) );
			self:error_exit( "get_kv_fail ret_1_not_string:" .. (err or "nil") .. table.tostring(ret_table_) );
		end

	else
		-- 返回的一定是一个table 否则是出错
		log.day_list_short( "dc_ssdb", "get_kv_fail", key, "ret_not_table" );
		self:error_exit( "get_kv_fail ret_not_table:" .. (err or "nil") );
	end

	ngx.exit(ngx.ERROR);  --严重出错
end



-------------------------------------------------------
_M.API_ssdb_save = function( self, ssdb_name, key, hash, value, err_call_back, ssdb_delay_, read_only_ )
	if  log.fast() then
		log.debug( "call ssdbMgr.API_ssdb_save, ssdb_name=" .. ssdb_name .. ", key=" .. key .. ", hash=" .. hash )
	end

	ngx.ctx.err_call_back = err_call_back;

	local key_hash_ = key .. "_h" .. hash

	--优先写入share cache
	local ret_ = self:set_shared_cache_ssdb( ssdb_name, key_hash_, value )
	if  ret_ and ssdb_delay_ and (tonumber(ssdb_delay_) == 1) then

		--log.day_list_short( "dc_ssdb", "dc_ssdb_delay", ssdb_name, key, hash )

		self:unlock_resty_lock( ssdb_name, key_hash_, read_only_ )
		self:set_get_set_tj( hash, "delay_ssdb" )
		return   --缓写
	end

	--lock 模式
	local lock_id_ = self:read_lock2( ssdb_name )
	if  lock_id_  == 0 then
		self:error_exit( "save_lock_err" );
		return
	end

	local instance = self:try_connect(ssdb_name);
	local ok, err = instance:hset(key, hash, value);

	self:try_close( instance, ssdb_name, err )
	self:read_unlock2(ssdb_name)

	--[[
	if  read_only_ then
		-- 扫描的只读请求不计算在内
	else
		-- 统计重复写入比例
		local save_dict_ = ngx_shared[ self.pre .. ssdb_name ]
		local save_flag_ = save_dict_:get( key_hash_ .. "_s" )
		if  save_flag_ then
			self:set_get_set_tj( hash, "set_re1" )
		else
			self:set_get_set_tj( hash, "set_re0" )
			save_dict_:set( key_hash_ .. "_s", 1, 60 )
		end
	end
	--]]

	if  not ok then
		log.day_list_short( "dc_ssdb",  "hset_fail", ssdb_name, key, hash, err );
		self:error_exit( "hset err:" .. (err or "nil") );
	else
		self:unlock_resty_lock( ssdb_name, key_hash_, read_only_ )
		--log.day_list_short( "dc_ssdb",  "hset_succ", ssdb_name, key, hash, #value );
	end

	self:set_get_set_tj( hash, "set_ssdb" )

	return "OK"
end



--- 进行ssdb key_hash锁检查
--- 业务一致性锁，相当于redis的远程锁
--- 同一个key-hash被锁的情况，实际上是非常稀少的
--- 这里复杂的检测逻辑不会对系统性能有较大影响
_M.check_ssdb_resty_lock = function( self, ssdb_name, key_hash_ )
	local dict_ = ngx_shared[ self.pre .. ssdb_name ]
	if  dict_ then
		--2秒中内循环试图锁定一个 key_hash_lk ( exists则等待 )
		local timeout_ = 2000   ---毫秒
		local step_    = 1      ---步长毫秒
		local cc_      = 0

		---计算锁id
		self._lock_id = self._lock_id + 1
		if  self._lock_id >= 100000 then self._lock_id = 1000 end

		while timeout_ > 0 do
			cc_ = cc_ + 1

			local ok, err = dict_:add( key_hash_ .. "_lk" , self._lock_id, 3)      ---锁住主键kh  12345_6_lk
			if  ok then
				if  cc_ > 10 then
					--循环等待其他锁的次数太多
					log.day_list_short( "dc_ssdb", "lock_rw", "lock_ok0", ngx.worker.id(), uu.get_act_cmd(),
							key_hash_, cc_, 2000-timeout_, step_, self._lock_id );
				end
				return self._lock_id   -- 成功
			else
				if  err == 'exists' then
					timeout_ = timeout_ - step_
					if  timeout_ <= 0 then
						log.day_list_short("dc_ssdb", "lock_rw", "error_time_out0",
								ngx.worker.id(), uu.get_act_cmd(), self.pre, ssdb_name,
								key_hash_, cc_, timeout_, step_ )

						---完全超时 read请求并没有发出去，也没有开始connect，不断开其他链接
						--self:error_exit( "error_time_out_lock" );
						--log.error( "error_time_out_lock" );
						--ngx.exit(ngx.ERROR);
						return 0, "error_time_out_lock"    --超时
					end

					--if  step_ < 3 then     --1 2 4 8
					--step_ = step_ * 2
					--end

					if  cc_ <= 10 then
						step_ = 50 + ( cc_ * 2)   ---等待时间逐步放大   500+2+4+6+8+10+12+14+16+18+20=500 + 110
					else
						step_ = (20-cc_) * 10      ---等待时间逐步缩小  100 90 80 70 60 50 40 30
						if  step_ < 10 then step_ = 10 end
					end

					--log.day_list_short( "dc_ssdb", "lock_rw", "step0", ngx.worker.id(), uu.get_act_cmd(),
						--key_hash_, timeout_, step_, debug.traceback() );
					ngx_sleep( step_ * 0.001 )

				else
					log.day_list_short( "dc_ssdb", "lock_rw", "error_step0",
							ngx.worker.id(), uu.get_act_cmd(), self.pre, ssdb_name,
							key_hash_, cc_, timeout_, step_, err );
					return  0, err  	---其他错误
				end
			end
		end

	else

		if  self.select_mod ~= 10 then
			---提示可能是百分表
			log.error( "error_select: plz check select_mod=" .. self.select_mod )
		end

		log.day_list_short("dc_ssdb", "lock_rw", "error_no_ssdb_name", ngx.worker.id(), uu.get_act_cmd(),
				self.pre, ssdb_name, key_hash_ );
		return 0, "error_no_ngx_shared"   ---找不到 ngx_shared
	end

	return 0, "unknown_err"   ---未知错误 (应该不能运行到这里，健壮性保证)
end



---远程调用解锁，只有解锁后，其他【进程或协程】的请求才能继续获得读写锁，进行读写操作
---和check_ssdb_resty_lock是一对
---read_only_ 不解锁，等待final统一解锁
_M.unlock_resty_lock = function( self, ssdb_name, key_hash_, read_only_ )
	if  read_only_ then
		return "ok"  ---不解锁
	end

	local dict_ = ngx_shared[ self.pre .. ssdb_name ]
	if  dict_ then
		local ok, err_ = dict_:delete( key_hash_ .. "_lk")
		if  ok then
			--删除成功
			--log.day_list_short( "dc_ssdb", "lock_rw", "unlock_ok", ngx.worker.id(), uu.get_act_cmd(), key_hash_ );
			return "ok"
		else
			log.day_list_short( "dc_ssdb", "lock_rw", "error_delete_unlock1", ngx.worker.id(),
					uu.get_act_cmd(), self.pre, ssdb_name, key_hash_, err_ );
			return "not_lock"
		end
	else
		log.error( "dict_not_exist=" .. self.pre .. ssdb_name )
		log.day_list_short( "dc_ssdb", "lock_rw", "error_dict_not_exist",  self.pre .. ssdb_name );
		return "no_dict"
	end
end


--- API 尝试获得远程锁
_M.API_ssdb_try_get_resty_lock = function(self, ssdb_name, key_hash_ )
	return self:check_ssdb_resty_lock(ssdb_name, key_hash_ )
end


--- API 解开远程锁 包装提供给外部调用
_M.API_ssdb_unlock_resty_lock = function( self, ssdb_name, key_hash_, read_only_ )
	return self:unlock_resty_lock( ssdb_name, key_hash_, read_only_ )
end


---读取一个数据
_M.API_ssdb_read = function( self, ssdb_name, key, hash, err_call_back, read_only_ )
	if  log.fast() then
		log.debug( "call ssdbMgr.API_ssdb_read, ssdb_name=" .. ssdb_name .. ", key=" .. key .. ", hash=" .. hash );
	end

	ngx.ctx.err_call_back = err_call_back;

	self:set_get_set_tj( hash, "get_call" )

	local key_hash_ = key .. "_h" .. hash

	--乐观锁r
	if  read_only_ then
		--只读接口
		--log.day_list_short( "dc_ssdb", "lock_rw", "read_only", ngx.worker.id(), uu.get_act_cmd(), key_hash_ );
	else
		local lock_id_, err_msg_ = self:check_ssdb_resty_lock( ssdb_name, key_hash_ )
		if  lock_id_ == 0 then
			self:error_exit( err_msg_ );
			ngx.exit(ngx.ERROR);
			return
		end
		ngx.ctx.__lock_id__ = lock_id_   ---记录lock_id
	end

	local ret2, stale_ = self:get_shared_cache_ssdb( ssdb_name, key_hash_ )
	if  ret2 then
		--log.debug( log.fast() and "ret2 cache hit." )
		if  stale_ then
			self:set_get_set_tj( hash, "get_cache_stale" )
		else
			self:set_get_set_tj( hash, "get_cache" )
		end
		return ret2   --cache命中
	end

	local ret = NGX_NULL;

	--lock 模式
	local lock_id_ = self:read_lock2(ssdb_name)
	if  lock_id_  == 0 then
		self:error_exit( "read_lock_err" );
		return
	end

	local instance = self:try_connect(ssdb_name);
	local ret_table_, err = instance:hget(key, hash);

	self:try_close( instance, ssdb_name, err )
	self:read_unlock2(ssdb_name)

	if  err then
		log.day_list_short( "dc_ssdb", "hget_fail|" .. key .. "|" .. hash  .. "|" .. (err or "nil") );
		self:error_exit( "hget err:" .. (err or "nil") );
		return;
	end

	uu.var_dump( ret_table_ );

	-- 处理返回值 返回是字符串或者NGX_NULL
	if  type(ret_table_) == 'table' then
		ret = ret_table_[1];
		if  ret and type(ret) == 'string' then
			if      NGX_NULL == ret then
				log.debug( log.fast() and "dc_ssdb ret not_found " .. key .. "_" .. hash );
				--log.day_list_short( "dc_ssdb", "hget_succ|" .. key .. "|" .. hash  .. "|empty" );
				--结果为空
			elseif  NGX_AUTH == ret then
				--没有权限(严重错误)
				self:error_exit( self.pre .. ssdb_name .. " ssdb " .. NGX_AUTH );
				ngx.exit(ngx.ERROR);
				return
			else
				--结果正常
				--log.day_list_short( "dc_ssdb", "hget_succ|" .. key .. "|" .. hash  .. "|" .. #ret );

				--select_, key_, str_
				self:set_get_set_tj( hash, "get_ssdb" )
				if  ngx.ctx.no_shared_cache == 'y' then
					--不保存 scan使用
					log.day_list_short( "dc_ssdb", "no_shared_cache", ssdb_name, key, hash );
				else
					if  #ret > 0 then
						self:set_shared_cache_ssdb( ssdb_name, key_hash_, ret )
					else
						log.day_list_simple( "dc_ssdb", "error_len0", ssdb_name, key, hash, ret );
					end
				end

			end
		else
			--SSDB错误 一般是超时
			log.day_list_short( "dc_ssdb", "hget_fail|" .. key .. "|" .. hash  .. "|ret_1_not_string|"
					.. table.tostring(ret_table_) );
			self:error_exit( "hget_fail err:" .. (err or "nil") .. table.tostring(ret_table_) );

			--严重出错一定退出
			ngx.exit(ngx.ERROR);
			return
		end

	else
		-- 返回的一定是一个table 否则是出错
		log.day_list_short( "dc_ssdb", "hget_fail|" .. key .. "|" .. hash  .. "|ret_not_table|" );
		self:error_exit( "hget err:" .. (err or "nil") );

		--严重出错一定退出
		ngx.exit(ngx.ERROR);
		return
	end

	return ret;
end



---用来校验not_found的读取
_M.read_check_not_found = function( self, ssdb_name, key, hash )

	local ret

	--lock 模式
	local lock_id_ = self:read_lock2(ssdb_name)
	if  lock_id_  == 0 then
		self:error_exit( "read_nil_lock_err" );
		return
	end

	local instance = self:try_connect(ssdb_name);
	local ret_table_, err = instance:hget(key, hash);

	self:try_close( instance, ssdb_name, err )
	self:read_unlock2(ssdb_name)

	if  err then
		log.day_list_short( "dc_ssdb", "hget_fail|" .. key .. "|" .. hash  .. "|" .. (err or "nil") );
		self:error_exit( "hget err:" .. (err or "nil") );
		return;
	end

	-- 处理返回值 返回是字符串或者NGX_NULL
	if  type(ret_table_) == 'table' then
		ret = ret_table_[1];
		if  ret and type(ret) == 'string' then
			return ret
		else
			--SSDB错误 一般是超时
			log.day_list_short( "dc_ssdb", "hget_fail|" .. key .. "|" .. hash  .. "|ret_1_not_string|"
					.. table.tostring(ret_table_) );
			self:error_exit( "hget_fail err:" .. (err or "nil") .. table.tostring(ret_table_) );

			--严重出错一定退出
			ngx.exit(ngx.ERROR);
			return
		end
	else
		-- 返回的一定是一个table 否则是出错
		log.day_list_short( "dc_ssdb", "hget_fail|" .. key .. "|" .. hash  .. "|ret_not_table|" );
		self:error_exit( "hget err:" .. (err or "nil") );

		--严重出错一定退出
		ngx.exit(ngx.ERROR);
		return
	end

	return ret;
end



-- multi hget
_M.API_ssdb_mread = function( self, ssdb_name, all_kv_list, err_call_back )

	if  log.fast() then
		log.debug( "call ssdbMgr.API_ssdb_mread, ssdb_name=" .. ssdb_name );
	end

	ngx.ctx.err_call_back = err_call_back;

	local fret = {};

	--先查cache
	local kv_list = {}           --需要ssdb的list
	for i=1, #all_kv_list do
		self.mread_tj.call = self.mread_tj.call + 1

		local key_hash_ = all_kv_list[i].key .. "_h" .. all_kv_list[i].hash
		local ret2, stale_ = self:get_shared_cache_ssdb( ssdb_name, key_hash_ )
		if  ret2 then
			--log.debug( log.fast() and "mread cache hit " .. key_hash_ )
			--log.day_list_short("dc_ssdb", "cache_hit_ssdb", "mm|cache_hit|" .. key_hash_ )
			fret[i] = ret2;
			if  stale_ then
				self.mread_tj.gets = self.mread_tj.gets + 1
			else
				self.mread_tj.get =  self.mread_tj.get + 1
			end

		else
			--log.debug( log.fast() and "mread need ssdb " .. key_hash_ )
			--log.day_list_short("dc_ssdb", "cache_hit_ssdb", "need_ssdb|" .. key_hash_ )
			--fret[i] = {}
			all_kv_list[i].__mi = i;     --mread index
			table_insert( kv_list, all_kv_list[i] )

			-- 只要触发流控，阻止请求ssdb，可以请求shared_cache
			if  self.read_only_ssdb_block > 0 then
				self.read_only_ssdb_block = self.read_only_ssdb_block - 1
				log.day_list_short( "dc_ssdb", "mread|read_only_ssdb_block|" .. self.read_only_ssdb_block );
				self:error_exit( "mread err:read_only_ssdb_block" .. self.read_only_ssdb_block,  "ssdb_block" );
				return;
			end

		end
	end


	local ret_list = {};

	--lock 模式
	local lock_id_ = self:read_lock2( ssdb_name )
	if  lock_id_  == 0 then
		self:error_exit( "mread_lock_err", "ssdb_block" );
		return
	end

	local instance = self:try_connect(ssdb_name);
	instance:init_pipeline();
	for i=1, #kv_list do
		instance:hget( kv_list[i].key, kv_list[i].hash );
	end
	local ret_table_list_, err = instance:commit_pipeline();

	self:try_close( instance, ssdb_name, err )
	self:read_unlock2(ssdb_name)

	if  err then
		log.day_list_short( "dc_ssdb", "mread|" .. (err or "nil") );
		self:error_exit( "mread err:" .. (err or "nil") );
		return;
	end

	--uu.var_dump( ret_table_list_ );

	if type(ret_table_list_) == 'table' then

		for i=1, #ret_table_list_ do
			local ret_table_ = ret_table_list_[i];
			local ret = NGX_NULL;

			local key_ = "nil";
			if  kv_list[i] then
				key_ = kv_list[i].key or "nil";
			end

			-- 处理返回值 返回是字符串或者NGX_NULL
			if  type(ret_table_) == 'table' then
				ret = ret_table_[1];
				if  ret and type(ret) == 'string' then
					if     NGX_NULL == ret then
						log.debug( log.fast() and "dc_ssdb ret not_found " .. key_ .. "_" .. kv_list[i].hash );
						--log.day_list_short( "dc_ssdb", "hget_succ|empty" );
					elseif NGX_AUTH == ret then
						--没有权限(严重错误)
						self:error_exit( self.pre .. ssdb_name .. " ssdb " .. NGX_AUTH );
						ngx.exit(ngx.ERROR);
						return
					else
						--log.day_list_short( "dc_ssdb", "hget_succ|" .. #ret );
						local key_hash_ = key_ .. "_h" .. kv_list[i].hash
						self:set_shared_cache_ssdb( ssdb_name, key_hash_, ret )
						self.mread_tj.set = self.mread_tj.set + 1
					end
				else
					ret = NGX_NULL;
					log.day_list_short( "dc_ssdb", "mread|ret_1_not_string|" .. key_ );
				end

			else
				-- 返回的一定是一个table 否则是出错
				ret = NGX_NULL;
				log.day_list_short( "dc_ssdb", "mread|ret_not_table|" .. key_ );
			end

			table_insert(ret_list, ret);
		end
	end


	--合并结果
	for i=1, #ret_list do
		if  kv_list[i] and kv_list[i].__mi then
			fret[ kv_list[i].__mi ] = ret_list[i]
		end
	end

	if  #fret ~= #all_kv_list then
		log.error("mread ret number not match")
		log.day_list_short( "dc_ssdb", "cache_hit_ssdb", "error_count")
	end

	return fret;

end



-- 删除一个key下的一个hash
_M.API_ssdb_del = function( self, ssdb_name, key, hash, err_call_back )

	log.debug( log.fast() and "call ssdbMgr.API_ssdb_del, ssdb_name="
			.. ssdb_name .. ", key=" .. key .. ", hash=" .. hash );
	ngx.ctx.err_call_back = err_call_back;

	--self, select_, key_, hash_ )
	self:clean_shared_cache_ssdb( ssdb_name, key, hash )

	--lock 模式
	local lock_id_ = self:read_lock2( ssdb_name )
	if  lock_id_  == 0 then
		self:error_exit( "del_lock_err" );
		return
	end

	local instance = self:try_connect(ssdb_name);
	local ret_table_, err = instance:hdel(key, hash);

	self:try_close( instance, ssdb_name, err )
	self:read_unlock2(ssdb_name)

	if  err then
		log.day_list_short( "dc_ssdb", "hdel_fail|" .. ssdb_name .. "|" .. key .. "|" .. hash  .. "|" .. (err or "nil") );
		self:error_exit( "hdel err:" .. (err or "nil") );
		return;
	end

	self:unlock_resty_lock( ssdb_name, key .. "_h" .. hash )  --key_hash_

	uu.var_dump( ret_table_ );

	if  ret_table_ and ret_table_[1] then
		return ret_table_[1];
	else
		return 0;
	end
end


-- 一个key下有多少个hash值
_M.API_ssdb_hlen = function( self, ssdb_name, key, err_call_back)

	log.debug( log.fast() and "call ssdbMgr.API_ssdb_hlen, ssdb_name=" .. ssdb_name .. ", key=" .. key );
	ngx.ctx.err_call_back = err_call_back;

	--lock 模式
	local lock_id_ = self:read_lock2( ssdb_name )
	if  lock_id_  == 0 then
		self:error_exit( "hlen_lock_err" );
		return
	end

	local instance = self:try_connect(ssdb_name);
	local ret_table_, err = instance:hsize(key);

	self:try_close( instance, ssdb_name, err )
	self:read_unlock2(ssdb_name)

	if  err then
		log.day_list_short( "dc_ssdb", "hlen_fail|" .. ssdb_name .. "|" .. key  .. "|" .. (err or "nil") );
		self:error_exit( "hget err:" .. (err or "nil") );
		return;
	end

	uu.var_dump( ret_table_ );

	if  ret_table_ and ret_table_[1] then
		return ret_table_[1];
	else
		return 0;
	end
end



--遍历一个连接下的所有值
_M.API_ssdb_hlist = function( self, ssdb_name, start_, end_, size_, err_call_back )
	log.debug( log.fast() and "call ns_ssdbMgr.API_ssdb_hlist, start=" .. start_ );
	ngx.ctx.err_call_back = err_call_back;

	--lock 模式
	local lock_id_ = self:read_lock2( ssdb_name )
	if  lock_id_  == 0 then
		self:error_exit( "hlist_lock_err" );
		return
	end

	local instance = self:try_connect(ssdb_name);
	local ret_table_, err = instance:hlist( start_, end_, size_ );

	self:try_close( instance, ssdb_name, err )
	self:read_unlock2(ssdb_name)

	if  err then
		log.day_list_short( "dc_ssdb", "hlist_fail|" .. ssdb_name .. "|" .. start_ .. "|" .. (err or "nil") );
		self:error_exit( "hget err:" .. (err or "nil") );
		return;
	end

	uu.var_dump( ret_table_ );
	return ret_table_ or {};
end


--获得一个key下的所有值
_M.API_ssdb_hget_all = function( self, ssdb_name, key, err_call_back )

	log.debug( log.fast() and "call ns_ssdbMgr.API_ssdb_hget_all, key=" .. key );
	ngx.ctx.err_call_back = err_call_back;

	--lock 模式
	local lock_id_ = self:read_lock2( ssdb_name )
	if  lock_id_  == 0 then
		self:error_exit( "hget_all_lock_err" );
		return
	end

	local instance = self:try_connect(ssdb_name);
	local ret_table_, err = instance:hgetall(key);

	self:try_close( instance, ssdb_name, err )
	self:read_unlock2(ssdb_name)

	if  err then
		log.day_list_short( "dc_ssdb", "hlen_fail|" .. ssdb_name .. "|" .. key .. "|" .. (err or "nil") );
		self:error_exit( "hget err:" .. (err or "nil") );
		return;
	end

	uu.var_dump( ret_table_ );
	return ret_table_ or {};
end


-- 出错处理
_M.error_exit = function(self, txt, trace_info_ )
	if  ngx.ctx.err_call_back then
		--log.error( "has set error_exit. " .. (txt or 'nil') );
		ngx.ctx.err_call_back(txt, trace_info_);
	else
		log.error( "not set error_exit. " .. (txt or 'nil'), trace_info_ );
	end
end



-- 关闭单个SSDB连接
_M.private_close = function( self, instance, ssdb_name_, shutdown )
	local ok, err;
	if  shutdown == true then
		ok, err = instance:close();
		log.day_list_short( "dc_ssdb",  "close|" .. ssdb_name_ )
		if  not ok then
			log.error("connection close: " .. (err or 'nil_err') .. ", ssdb_name=" .. ssdb_name_ );
		else
			log.debug( log.fast() and "connection close ok. ssdb_name=" .. ssdb_name_ );
		end
	else
		ok, err = instance:set_keepalive( 120000, 128 );   --2min
		if  not ok then
			log.error("connection keepalive: " .. (err or 'nil_err') .. ", ssdb_name=" .. ssdb_name_ );
		else
			if  log.fast() then
				log.debug( "connection keep alive ok. ssdb_name=" .. ssdb_name_ );
			end
		end
	end
end



-- 按连接名字获得ssdb配置
_M.get_ssdb_config = function( self, ssdb_name_ )
	if  self.config_cache[ ssdb_name_ ]  then
		--先查config cache
		if  log.fast() then
			log.debug( "find config_cache for " .. ssdb_name_ )
		end
		return  self.config_cache[ ssdb_name_ ];
	else
		--if  ns_config and ns_config.ssdb_db_select_s2 and ns_config.ssdb_server_s2 then
		--config_server = config_server_;
		--config_select = config_select_;
		if  self.config_server and self.config_select then
			if  self.config_select[ssdb_name_] then
				local ret_ = self.config_server[ self.config_select[ssdb_name_] ];
				if  ret_ then
					log.debug( log.fast() and "set config_cache for " .. ssdb_name_ );
					self.config_cache[ ssdb_name_ ] = ret_;
					return ret_;
				end
			end
		end
	end

	self:error_exit( 'error ssdb_name = ' .. (ssdb_name_ or 'nil') );
end


-- 按照select_name_, 获得SSDB连接
-- ssdb_name_ = comm, uin,
_M.try_connect = function( self, ssdb_name_ )
	local config_ = self:get_ssdb_config( ssdb_name_ );

	--新建连接
	if  log.fast() then
		log.debug("create new ssdb connection. ssdb_name=" .. ssdb_name_ );
	end

	local instance = self:private_try_connect( config_.ip, config_.port, config_.auth, ssdb_name_ );
	return instance;
end


-- 尝试关闭一个ssdb连接
_M.try_close = function( self, instance, ssdb_name_, err )
	if  err then
		log.error("try_close: " .. uu.to_str(err) .. ", ssdb_name=" .. ssdb_name_ )
		self:private_close( instance, ssdb_name_, true )   --shutdown socket
	else
		self:private_close( instance, ssdb_name_ )         --keep alive socket
	end
end


-----------------------------------------------------------------------------utils---------------------
-- config = { ip, port, auth, select }
-- 建立ssdb连接
_M.private_try_connect = function( self, ip, port, auth, ssdb_name_ )

	local instance = ns_ssdb:new();
	instance:set_timeout(3000);      --3s

	local ok, err = instance:connect( ip, port );
	if  not ok then
		self:read_unlock2(ssdb_name_)   		--- 唤醒下一个sema
		self:error_exit( "fail connect. msg=" .. err );
		return
	end

	--log.day_list( "socket_ssdb", ngx.worker.id(), ip, port,
			--tostring( instance ), tostring( instance.sock ),
			--instance:get_reused_times(), uu.to_str(ok), uu.to_str(err) )

	if  auth and #auth>0 then
		local count, err = instance:get_reused_times();
		if  0 == count then
			--set auth
			ok, err = instance:auth( auth );
			if  not ok then
				self:read_unlock2(ssdb_name_)  	 --- 唤醒下一个sema
				self:error_exit( "fail auth. msg=" .. err );
				return;
			end

		elseif err then
			self:read_unlock2(ssdb_name_)  	 --- 唤醒下一个sema
			self:error_exit( "fail get_reused_times. msg=" .. err );
			return;
		else
			if  log.fast() then
				log.debug( "get_reused_times ssdb=" .. count );
			end
		end
	end

	return instance;
end


----------------------------------------------------------------------------

--写校验[_k_] [_h_]
_M.private_check_set_key = function(self, db_name_, key, hash, value )
	if  value._k_ then
		if  ( tonumber(value._k_) and ( tonumber(key) == tonumber(value._k_) ) ) or (key == value._k_) then
			--ok
		else
			log.day_list_short( "dc_ssdb", "error_key_check", "set", db_name_, key, value._k_, hash, debug.traceback() );
			log.error( "error_set_key_check" )
			return false;
		end
	else
		value._k_ = tonumber(key) or key;  --首次设置
	end

	if  value._h_ then
		if  ( tonumber(value._h_) and ( tonumber(hash) == tonumber(value._h_) ) ) or (hash == value._h_) then
			--ok
		else
			log.day_list_short( "dc_ssdb", "error_hash_check", "set", db_name_, key, value._h_, hash );
			log.error( "error_set_hash_check" )
			return false;
		end
	else
		if  tonumber(hash) then
			value._h_ = tonumber(hash) --首次设置
		end
	end

	value._t_ = uu.now()    --最后一次写入时间
	return true;
end



--读校验[_k_] [_h_]
_M.private_check_get_key = function(self, db_name_, key, hash, value )
	if  value._k_ then
		if  ( tonumber(value._k_) and ( tonumber(key) == tonumber(value._k_) ) ) or (key == value._k_) then
			--ok
		else
			--读取出错
			log.day_list_short( "dc_ssdb", "error_key_check", "get", db_name_, key, value._k_, hash, debug.traceback() );
			log.error( "error_get_key_check" )
			return false;
		end
	end

	if  value._h_ then
		if  ( tonumber(value._h_) and ( tonumber(hash) == tonumber(value._h_) ) ) or (hash == value._h_) then
			--ok
		else
			--读取出错
			log.day_list_short( "dc_ssdb", "error_hash_check", "get", db_name_, key, value._h_, hash );
			log.error( "error_get_hash_check" )
			return false;
		end
	end

	return true;
end




-----------------------------ngx_shared--------------------------------




--设置全局数据缓存kv 十分表
_M.set_shared_cache_ssdb = function( self, select_, key_, str_, expire_ )
	if  select_ and key_ and str_  then

		local dict_name_
		if  type(select_) == 'number' then
			dict_name_ = self.pre .. (select_ % self.select_mod)
		else
			dict_name_ = self.pre ..  select_
		end

		local share_ = ngx_shared[ dict_name_ ];
		if  share_ then

			if  expire_ and type(expire_) == 'string' then
				--log.debug( log.fast() and "expire_=" .. expire_ .. " k=" .. key_ .. " v=" .. str_ )
				expire_ = tonumber(expire_) or 0
			end

			if  expire_ and expire_ >=0 and expire_ < 7200 then
				--normal
			else
				expire_ = 7200     --2*60*60
			end

			local ok, err, forcible = share_:set( key_, str_, expire_ );
			if  ok then
				if  forcible then
				    log.day_list_short("dc_ssdb", "cache_hit_ssdb", "set_forcible", dict_name_, key_, err )
				end
				return true
			else
				if  err then
					--share_:delete( key_ )
					--set出错，立即删除缓存
					local ok, err, forcible = share_:set( key_, nil );
					if  err or forcible then
						log.day_list_short("dc_ssdb", "cache_hit_ssdb", "error_delete", dict_name_, key_, err, forcible )
					end
				end
				log.day_list_short("dc_ssdb", "cache_hit_ssdb", "error_set", dict_name_, key_, err )
			end
		else
			log.day_list_short("dc_ssdb", "cache_hit_ssdb", "error_set_no_dict", dict_name_, key_)
		end
	end

	return false   --写入cache失败
end



--获得全局数据缓存kv 十分表
_M.get_shared_cache_ssdb = function( self, select_, key_ )
	if  select_ and key_ then

		local dict_name_
		if  type(select_) == 'number' then
			dict_name_ = self.pre .. (select_ % self.select_mod)
		else
			dict_name_ = self.pre ..  select_
		end

		local share_ = ngx_shared[ dict_name_ ];
		if  share_ then

			local str_, f_, stale_ = share_:get_stale( key_);
			--uu.report_share_free_space( share_, dict_name_, key_, "dc_ssdb" )

			if  str_ then
				return str_, stale_  	--正常返回
			else
				--有报错的情况 ret = nil, error
				if  f_ then
					log.day_list_short("dc_ssdb", "cache_hit_ssdb", "error_get_nil", dict_name_, key_, f_ )
				end
			end
		else
			log.day_list_short("dc_ssdb", "cache_hit_ssdb", "error_get_no_dict", dict_name_, key_)
		end
	end

	return nil, nil
end


--清理掉一个cache数据 十分表
_M.clean_shared_cache_ssdb = function(self, select_, key_, hash_ )
	if  select_ and key_ and hash_ then
		local dict_name_
		if  type(select_) == 'number' then
			dict_name_ = self.pre .. (select_ % self.select_mod)
		else
			dict_name_ = self.pre ..  select_
		end

		local share_ = ngx_shared[ dict_name_ ];
		if  share_ then
			local ok, err, forcible = share_:set( key_ .. "_h" .. hash_ , nil );
			if  err or forcible then
				log.day_list_short("dc_ssdb", "cache_hit_ssdb", "error_delete", dict_name_, key_, err, forcible )
			end
		end
	end
end


--- API 清理掉一个cache数据
_M.API_ssdb_clean_shared_cache = function(self, p_select_, key_, hash_ )
	self:clean_shared_cache_ssdb(p_select_, key_, hash_ )
end


--- 同进程信号锁semaphore lock
--- 相同worker下的不同协程同步 避免使用同一个socket句柄收发消息
--- limit_  流控系数，  ---优先保护写
--- return 0=未拿到锁(超限制 timeout)  1=拿到锁
_M.read_lock2 = function( self, select_ )

	if  not self.sema_list[ select_ ] then
		--sema=信号锁  t=当前秒  link=当前秒的赛道并行数 max=当前秒出现的赛道最大值
		self.sema_list[ select_ ] = { sema = ns_semaphore.new( __SEMA_RESOURCES__ ), t=0, link=0, max=0 }
	end

	local sema_ = self.sema_list[ select_ ].sema

	if  sema_:count() < __SEMA_PROTECT_BREAK__ then
		-- 资源数耗尽,队列过长
		log.day_list_short( "dc_ssdb_cost", ngx.worker.id(), "wait_list_limit",
				self.pre, select_, sema_:count() )

		self.read_only_ssdb_block = __BUSY_BLOCK_LIMIT__   ---触发限流
		return 0
	end

	local start_  = ngx.now()
	local ok, err = sema_:wait(2)   --最大等待2秒
	local cost_   = ngx.now() - start_
	if  ok then
		--ngx.sleep( 0.001 * math.random( 200, 400 ) )    --haima test

		-- link +1
		self.sema_list[ select_ ].link = self.sema_list[ select_ ].link + 1
		if  self.sema_list[ select_ ].max < self.sema_list[ select_ ].link then
			self.sema_list[ select_ ].max = self.sema_list[ select_ ].link
		end

		if  cost_ >= 0.01 then   --10ms
			log.day_list_short( "dc_ssdb_cost", ngx.worker.id(), "wait_ok", self.pre, select_,
					math.floor(cost_*1000),
					'sema=' .. sema_:count(),
					'max='..self.sema_list[ select_ ].max,
					'link='..self.sema_list[ select_ ].link )
		end

		return 1  -- 正常唤醒 拿到锁
	else
		-- 一般都是 timeout
		log.day_list_short( "dc_ssdb_cost", ngx.worker.id(), "wait_timeout", self.pre, select_,
				err,  math.floor(cost_*1000),
				'sema=' .. sema_:count(),
				'max='..self.sema_list[ select_ ].max,
				'link='..self.sema_list[ select_ ].link )

		log.error("ERROR: read_semaphore " .. err)
		return 0   -- 未拿到锁
	end

end



---信号锁解锁 semaphore unlock， 和read_lock2是一对
_M.read_unlock2 = function( self, select_ )
	if  self.sema_list and self.sema_list[ select_ ] then
		local sema_info_ = self.sema_list[ select_ ]

		-- thread减1
		self.sema_list[ select_ ].link = self.sema_list[ select_ ].link - 1

		-- 转秒计数清0
		if  sema_info_.t < ngx.time() then
			sema_info_.t = ngx.time()

			if  sema_info_.link ~= 0 then
				log.day_list_short( "dc_ssdb_cost", ngx.worker.id(), "sema_max", self.pre, select_,
						'sema=' .. sema_info_.sema:count(),
						'max='..self.sema_list[ select_ ].max,
						'link='..self.sema_list[ select_ ].link )
			end
			sema_info_.max = 0
		end

		if  sema_info_.sema:count() < __SEMA_RESOURCES__ then
			sema_info_.sema:post(1)   --可用资源数+1
		else
			-- 超限(不用拉起) 应该不会发生
			log.day_list_short( "dc_ssdb_cost", ngx.worker.id(), "error_sema_bigger_res", self.pre, select_,
					sema_info_.sema:count() )
		end
	end
end



return _M;
